import pandas as pd
from es_pandas import es_pandas


def write_to_es(file_path, es_hosts, index_name="aclr_weather_hour"):
    df = pd.read_csv(file_path)
    if df.empty:
        raise Exception("没有数据需要导入")
    csv_columns = [
        "region_code",
        "datetime",
        "tmp",
        "rh",
        "wind_s",
        "wind_d",
        "pre",
        "prs",
    ]
    if set(csv_columns) - set(df.columns):
        raise Exception("缺少列:{}".format(set(csv_columns) - set(df.columns)))
    df = df[csv_columns]
    df.fillna(0, inplace=True)
    if len(df) == 0:
        raise Exception("文件为空，导入失败")
    df.drop_duplicates(inplace=True)
    # ep = es_pandas(hosts='zxtech:Zxod112_shining10@192.168.80.224:19200')
    ep = es_pandas(hosts=es_hosts)
    df["datetime"] = pd.to_datetime(df["datetime"], format="%Y%m%d%H")
    # df = df.astype({'datetime': 'datetime64[ns]'})
    df = df.astype({"region_code": str})
    ep.to_es(df, index_name)


if __name__ == "__main__":
    csv_path = "/root/河北省.csv"
    es_host = "zxtech:Zxod112_shining10@192.168.83.22:19200"
    index_name = "aclr_weather_hour"
    # es_host = 'zxtech:Zxod112_shining10@192.168.81.192:9200'
    write_to_es(csv_path, es_host, index_name)
